package cn.azzhu.day06;

import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.api.java.utils.ParameterTool;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
import redis.clients.jedis.Jedis;

/**
 * @author azzhu
 * @create 2020-09-20 12:19:46
 */
public class MyRedisSink extends RichSinkFunction<Tuple3<String,String,String>> {
    private transient Jedis jedis;
    @Override
    public void open(Configuration parameters) throws Exception {
        ParameterTool params =(ParameterTool) getRuntimeContext().getExecutionConfig().getGlobalJobParameters();
        jedis = new Jedis(params.getRequired("redis.host"), 6379, 5000);
        jedis.select(params.getInt("redis.db",0));
    }

    @Override
    public void invoke(Tuple3<String, String, String> value, Context context) throws Exception {
        if(!jedis.isConnected()) {
            jedis.connect();
        }
        jedis.hset(value.f0, value.f1, value.f2);
    }

    @Override
    public void close() throws Exception {
        jedis.close();
    }
}
